package com.sptus.framework.mqtt;

import java.io.InputStream;

import com.sptus.common.utils.StringUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import com.alibaba.fastjson2.JSONObject;
import com.sptus.framework.kafka.KafkaProducerService;
import com.sptus.framework.mapper.util.SslUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * Mqtt客户端类
 * @author Mr.Liu
 * @date 2021/3/31 17:58
 */
@Slf4j
public class MyMqttClient {
    /** mqtt客户端对象 */
    public   static MqttConnectOptions mqttConnectOptions;
    /** MQTT连接对象 对连接进行设置 */
    public static MqttClient mqttClient;
    private MemoryPersistence memoryPersistence;
    /** 定义需要的属性变量 */
    private String serviceUrl;
    private String userName;
    private String password;
    private String clientId;

    private  InputStream trustStorePath;

    private String trustStorePassword;

    private InputStream keyStorePath;

    private String keyStorePassword;

    private  MqttCallback mqttcallback;
    /**
     * 定义构造函数
     *
     * @param serviceUrl
     * @param userName
     * @param password
     * @param trustStorePath
     * @param trustStorePassword
     * @param keyStorePath
     * @param keyStorePassword
     */
    public MyMqttClient(String serviceUrl, String userName, String password, String clientId, String trustStorePath, String trustStorePassword, String keyStorePath, String keyStorePassword,MqttCallback mqttcallback) {
        this.serviceUrl = serviceUrl;
        this.userName = userName;
        this.password = password;
        this.clientId = clientId;
        this.trustStorePath = MyMqttClient.class.getClassLoader().getResourceAsStream("iot/truststore.jks");
        this.trustStorePassword=trustStorePassword;
        this.keyStorePassword=keyStorePassword;
        this.keyStorePath = MyMqttClient.class.getClassLoader().getResourceAsStream("iot/cert.jks");
        this.mqttcallback=mqttcallback;
    }
    public MyMqttClient() {

    }

    public static void main(String[] args) {
        System.out.println(System.getProperty("user.dir"));
    }
    /**
     * 类初始化，建立mqttClient连接
     */
    public void init() throws Exception {
        try {
            // 初始化连接对象
            mqttConnectOptions = new MqttConnectOptions();
            
            if (mqttConnectOptions != null) {
                // 设置连接参数
                mqttConnectOptions.setCleanSession(false); // 保持会话状态
                mqttConnectOptions.setConnectionTimeout(30); // 连接超时30秒
                mqttConnectOptions.setMqttVersion(4); // 使用MQTT 3.1.1版本
                mqttConnectOptions.setUserName(userName);
                mqttConnectOptions.setPassword(password.toCharArray());
                
                // SSL配置
                if (trustStorePath != null && keyStorePath != null) {
                    mqttConnectOptions.setSocketFactory(
                        SslUtil.getSocketFactory(trustStorePath, trustStorePassword, keyStorePath, keyStorePassword)
                    );
                }
                
                // 设置心跳和重连参数
                mqttConnectOptions.setKeepAliveInterval(60); // 心跳间隔60秒
                mqttConnectOptions.setAutomaticReconnect(true); // 启用自动重连
                mqttConnectOptions.setMaxReconnectDelay(30000); // 最大重连延迟30秒
                
                // 初始化持久化
                memoryPersistence = new MemoryPersistence();
                
                if (clientId != null) {
                    try {
                        mqttClient = new MqttClient(serviceUrl, clientId, memoryPersistence);
                        log.info("MQTT客户端创建成功，clientId: {}", clientId);
                    } catch (MqttException e) {
                        log.error("创建MQTT客户端失败: {}", e.getMessage(), e);
                        throw e;
                    }
                } else {
                    throw new IllegalArgumentException("clientId不能为空");
                }
            } else {
                throw new IllegalStateException("mqttConnectOptions初始化失败");
            }
            
            if (mqttClient != null) {
                // 设置回调函数
                if (ObjectUtils.isEmpty(this.mqttcallback)) {
                    mqttClient.setCallback(new MqttReceiveCallback());
                } else {
                    mqttClient.setCallback(this.mqttcallback);
                }
                
                log.info("开始连接MQTT服务器: {}", serviceUrl);
                mqttClient.connect(mqttConnectOptions);
                
                if (mqttClient.isConnected()) {
                    log.info("MQTT连接成功，clientId: {}", clientId);
                } else {
                    throw new MqttException(MqttException.REASON_CODE_CONNECT_IN_PROGRESS);
                }
            } else {
                throw new IllegalStateException("mqttClient对象为空，连接失败");
            }
            
        } catch (Exception e) {
            log.error("MQTT初始化失败: {}", e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 订阅主题
     * @param topic 主题
     * @param qos 消息质量
     */
    public static void subscribeTopic(String topic, int qos){
        if (mqttClient != null && mqttClient.isConnected() && topic != null){
            try {

                mqttClient.subscribe(topic, qos);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }else {
            System.out.println("订阅主题失败！mqttClient == null || mqttClient.isConnected() == false");
        }
    }

    /**
     * 取消订阅主题
     * @param topic 主题名称
     */
    public void cleanTopic(String topic){
        if (mqttClient != null && mqttClient.isConnected()){
            try {
                mqttClient.unsubscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }else {
            System.out.println("取消订阅失败！");
        }
    }

    /**
     * 发布消息
     * @param publishTopic 发布消息的主题名称
     * @param message 消息内容
     * @param qos 消息质量
     */
    public static void publishMessageMqtt(String publishTopic, String message, int qos){
        // 判断是否连接
        if (mqttClient != null && mqttClient.isConnected()){
            System.out.println("发布消息.......");
            System.out.println(StringUtils.format("发布消息人的clientId = {} topic = {}  message = {} ",mqttClient.getClientId(),publishTopic,message));
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(qos);
            mqttMessage.setPayload(message.getBytes());
            MqttTopic mqttTopic = mqttClient.getTopic(publishTopic);
            if (mqttTopic != null){
                try {
                    MqttDeliveryToken publish = mqttTopic.publish(mqttMessage);
                    if (!publish.isComplete()){
                        System.out.println("消息发布成功！");
                    }
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }else {
                System.out.println("mqttTopic == null");
            }
        }else {
            System.out.println("mqttClient == null || mqttClient.isConnected() == false");
        }
    }
    
    public static final String TRANSFER_TOPIC_KAFKA = "transfer-topic";
    /**
     * 发布消息
     * @param publishTopic 发布消息的主题名称
     * @param message 消息内容
     * @param qos 消息质量
     */
    public static void publishMessage(String publishTopic, String message, int qos){
        publishMessageMqtt(publishTopic, message, qos);
//            log.info("发布消息.......publishTopic:{} message:{}",publishTopic,message);
//            try {
//            	/*String[] split = publishTopic.split("/");
//                String key = split.length > 1 ? split[1] : "";
//                if(StringUtils.isEmpty(key)) {
//                	 KafkaProducerService.sendMessage(TOPIC_KAFKA, message);
//                }else {
//                	 KafkaProducerService.sendMessage(TOPIC_KAFKA, key, message);
//                }*/
//            	JSONObject payload = new JSONObject();
//                payload.put("publishTopic", publishTopic);
//                payload.put("message", message);
//                payload.put("qos", qos);
//                KafkaProducerService.sendMessage(TRANSFER_TOPIC_KAFKA, payload.toString());
//                log.info("✅ 已发送到 Kafka：" + payload.toJSONString());
//            } catch (Exception e) {
//            	log.error("❌ Kafka 发送失败：" + e.getMessage());
//            }
    }

    // 添加连接状态检查方法
    public static boolean isConnected() {
        return mqttClient != null && mqttClient.isConnected();
    }

    // 添加重连方法
    public static void reconnect() throws Exception {
        if (mqttClient != null && mqttConnectOptions != null) {
            try {
                if (!mqttClient.isConnected()) {
                    log.info("尝试重新连接MQTT服务器...");
                    mqttClient.connect(mqttConnectOptions);
                    
                    if (mqttClient.isConnected()) {
                        log.info("MQTT重连成功，重新订阅主题");
                        mqttClient.subscribe("/#", 0);
                    }
                }
            } catch (MqttException e) {
                log.error("MQTT重连失败: {}", e.getMessage(), e);
                throw e;
            }
        }
    }
}

